package com.tidu.strategy;

import com.tidu.strategy.abstractSync.AbstractSyncStrategy;
import com.tidu.strategy.sync.SyncStrategy;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import javax.sql.DataSource;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.stream.Collectors;

@Slf4j
public class MultiTableSyncStrategy extends AbstractSyncStrategy {

    private final List<String> tableNames;
    private final List<SingleTableSyncStrategy> strategies = new ArrayList<>();
    private final Map<String, String> progressMap = new ConcurrentHashMap<>();
    private final Executor mutilTableSyncExecutor;

    public MultiTableSyncStrategy(
            List<String> tableNames,
            DataSource sourceDataSource,
            DataSource targetDataSource) {
        super(sourceDataSource, targetDataSource);
        this.tableNames = tableNames;
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        // 1.核心线程池大小
        executor.setCorePoolSize(16);
        // 2.最大线程池大小
        executor.setMaxPoolSize(32);
        // 3.队列大小
        executor.setQueueCapacity(99999);
        // 4.线程名称
        executor.setThreadNamePrefix("multi-table-sync-");
        // 5.拒绝策略
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
        executor.initialize();
        this.mutilTableSyncExecutor = executor;

        this.initStrategies(sourceDataSource, targetDataSource);
    }

    private void initStrategies(DataSource sourceDs, DataSource targetDs) {
        for (String tableName : tableNames) {
            strategies.add(new SingleTableSyncStrategy(tableName, sourceDs, targetDs));
        }
    }

    /**
     * 启动多线程并发同步任务
     */
    @Override
    public List<Map<String, Object>> fetchData() {
        CountDownLatch latch = new CountDownLatch(strategies.size());
        strategies.forEach(strategy ->
                CompletableFuture.runAsync(() -> {
                    try {
                        List<Map<String, Object>> records = strategy.fetchData();
                        strategy.writeData(records);
                        progressMap.put(strategy.getTableName(), strategy.getProgress());
                    } catch (Exception e) {
                        log.error("Error syncing table: {}", strategy.getTableName(), e);
                    } finally {
                        latch.countDown();
                    }
                }, mutilTableSyncExecutor)
        );
        try {
            latch.await();
        } catch (Exception e) {
            log.error("Error waiting for all tasks to complete.", e);
        }
        return Collections.emptyList(); // 不再汇总数据
    }

    /**
     * writeData 不再使用（由 fetchData 中直接调用）
     */
    @Override
    public void writeData(List<Map<String, Object>> records) {
        // 由 fetchData 中并发调用
    }

    /**
     * 判断是否全部表同步完成
     */
    @Override
    public boolean isCompleted() {
        return strategies.stream().allMatch(SyncStrategy::isCompleted);
    }

    /**
     * 获取同步进度（按表统计）
     */
    @Override
    public String getProgress() {
        return progressMap.entrySet().stream()
                .map(e -> e.getKey() + ": " + e.getValue())
                .collect(Collectors.joining("\n"));
    }

}
